# -*- coding: utf-8 -*-
# @Time    : 2019/10/29 20:46
# @Author  : Damn7Kx
# @Software: PyCharm
import kafka
import json
import time,random

class Kafka_(object):
    '''
    单例模式使每个进程在调用该类时,
    只初始化一次kafka对象,
    防止实例化太多,kafka关闭连接,导致的数据插入失败的问题
    '''

    _producer = None
    __init_flag = True
    def __new__(cls, *args, **kwargs):
        if cls._producer is None:
            cls._producer = kafka.KafkaProducer(
                bootstrap_servers=['127.0.0.1:9092', '127.0.0.2:9092', '127.0.0.3:9092'])
        return cls._producer

    def __init__(self):
        if Kafka_.__init_flag:
            print('这是初始化Kafka.....')
            Kafka_.__init_flag = False


if __name__ == '__main__':
    try:
        K = Kafka_()
        # 这里item是item类型,将其转为字典
        item = {}
        data = json.dumps(dict(item), ensure_ascii=False)
        if item['formats'] == "weibo":
            K.send('weibotopic', data.encode(), key=b'weibo',
                   partition=random.choice([0, 1, 2]))
            print('weibotopic发送数据============', data)
            time.sleep(0.5)
        elif item['formats'] == "tieba":
            K.send('tiebatopic', data.encode(), key=b'tieba',
                   partition=random.choice([0, 1, 2]))
            print('tiebatopic发送数据============', data)
            time.sleep(0.5)
        else:
            K.send('metaserchtopic', data.encode(), key=b'metasearch',
                   partition=random.choice([0, 1, 2]))
            print('metasearchtopic发送数据============', data)
            time.sleep(0.5)
    except Exception as e:
        print('Kafka管道文件发生错误:', e)